#include "myserver.h"
#include <stdio.h>
#include "mainconfig.h"
#include <time.h>
#include <unistd.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>
#include "processmessage.h"


my_server  * pmyserver;

my_server *  init_myserver()
{
	my_server * pserver=malloc(sizeof(my_server));
	
	if(NULL  == pserver)
	{
		perror("malloc");
		return NULL;
	}
	pmyserver=pserver;
	memset(pserver,0,sizeof(my_server));

	int sin_size=sizeof(struct sockaddr_in);       	
	if((pserver->serversock_fd=socket(AF_INET,SOCK_STREAM,0))==-1)
	{
		perror("socket");
		return NULL;
	 }

	pserver->server_addr.sin_family=AF_INET;
	pserver->server_addr.sin_port=htons(SERVPORT);
	pserver->server_addr.sin_addr.s_addr=INADDR_ANY;
	bzero(&(pserver->server_addr.sin_zero),8);


	pserver->stop_flag=0;

	memset(&pserver->sock_queue,0,sizeof(struct sock_queuen_node)*SOCK_QUEUE_LEN);

	pthread_mutex_init(&pserver->sq_mutex,0);
	pthread_cond_init(&pserver->sq_cond,0);
	pthread_mutex_init(&pserver->workercnt_mutex,0);

	return pserver;
}

int close_myserver(my_server * pserver)
{
	if(NULL==pserver)
	return -1;
	else
	{
		while(pserver->stop_flag!=2)
		{
			sleep(10);
		}
		close(pserver->serversock_fd);
		free(pserver);
		printf("close");
		return 0;

	}
}
void signal_handller(int signum)
{
	pmyserver->stop_flag=signum;
}

int start_myserver(my_server * pserver)
{
	int ret=0;
	int i=0;
	pthread_t monitor_thread_id;
	pthread_t worker_thread_id;

	if(bind(pserver->serversock_fd,(struct sockaddr*)&pserver->server_addr,sizeof(struct sockaddr))==-1)
	{
		printf("bind failed\n");
		return -1;
	}
	if(listen(pserver->serversock_fd,BACKLOG)==-1)
	{
		printf("listen failed\n");
		return -1;
	}
	printf("listening......\n");
	
	signal(SIGTERM,signal_handller);
	signal(SIGINT,signal_handller);


	pthread_create(&monitor_thread_id,NULL,monitor_Thread,pserver);
	for(i=0;i<NUMBER_WOKRER_THEAD;i++)
	{
		ret=pthread_create(&worker_thread_id,NULL,worker_thread,pserver);
		if(ret==0)
		{
			pthread_mutex_lock(&pserver->workercnt_mutex);
			pserver->num_workerThread++;
			pthread_mutex_unlock(&pserver->workercnt_mutex);
		}
		else
		{
			perror("pthread_create");
		}
	}
	return 0;
}

int store_new_connect_queue(int sock_fd,my_server * pserver,struct sockaddr * aceptsock)
{
	pthread_mutex_lock(&pserver->sq_mutex);
	if(pserver->sq_head-pserver->sq_tail>=SOCK_QUEUE_LEN-1)
	{
		sleep(1); 
	}
	else
	{
		pserver->sock_queue[pserver->sq_head].accept_addr= * aceptsock;
		pserver->sock_queue[pserver->sq_head].accepted_sock_fd=sock_fd;
		pserver->sq_head++;
		pserver->sq_head = pserver->sq_head%SOCK_QUEUE_LEN;
		pthread_cond_signal(&pserver->sq_cond);
	}
	pthread_mutex_unlock(&pserver->sq_mutex);
	return 0;
}


int add_new_connection(my_server *pserver)
{
	int addrlen=sizeof(struct sockaddr);
	struct sockaddr  accept_sock;
	int sock =accept(pserver->serversock_fd ,&accept_sock,&addrlen);
        if(sock<0)
	{
		return -1;
	}
	else
	{
		store_new_connect_queue(sock, pserver,&accept_sock);
		return 0;
	}	
}


int monitor_Thread(my_server * pserver)
{
	fd_set read_set;
	FD_ZERO(&read_set);
	FD_SET(pserver->serversock_fd,&read_set);
	struct timeval tv;
	tv.tv_sec=1;
	tv.tv_usec=0;
	int maxfd=pserver->serversock_fd;
	printf("monitor_thread start ......\n");

	while(pserver->stop_flag==0)
	{
	//if the last para of select is null we can get the connection
		if(select(maxfd+1,&read_set,NULL,NULL,NULL)!=-1)
		{
			if(FD_ISSET(pserver->serversock_fd,&read_set))
			{
				add_new_connection(pserver);
				printf("add one connection");
			}
		}
	}
	
	//wait for worker thread to stop
	while(pserver->num_workerThread!=0)
	{
		sleep(10);
	}
	pserver->stop_flag=2;
	return 0;
}

int getValidsock(my_server * pserver ,struct sock_queuen_node * psq )
{
	pthread_mutex_lock(&pserver->sq_mutex);
	pthread_cond_wait(&pserver->sq_cond,&pserver->sq_mutex);
	memcpy(psq,&pserver->sock_queue[pserver->sq_tail],sizeof(struct sock_queuen_node));
	pserver->sq_tail++;
	pserver->sq_tail =pserver->sq_tail %SOCK_QUEUE_LEN;
	pthread_mutex_unlock(&pserver->sq_mutex);
	return 0;

}

int worker_thread(my_server  * pserver)
{
	struct sock_queuen_node  sq;
	int rcvlen=0;

	char * prcvbuf =calloc(1,RECV_BUF_SIZE);
	if(NULL== prcvbuf)
	{
		return -1;
	}
	printf("worker_thread start ......\n");

	while(pserver->stop_flag==0)
	{
		int ret= getValidsock(pserver,&sq);
		if(-1==ret)
		{
			sleep(1);	
		}
		else
		{
			rcvlen = recv(sq.accepted_sock_fd,prcvbuf,RECV_BUF_SIZE,0);
			if(rcvlen==-1)
			{
				perror("recv");
			}
			else
			{
				prcvbuf[rcvlen]='\0';
				printf("recv:\n%s\n",prcvbuf);
				processmessage(sq.accepted_sock_fd,prcvbuf,rcvlen);
			}	
		}

	}

	free(prcvbuf);
	pthread_mutex_lock(&pserver->workercnt_mutex);
	pserver->num_workerThread--;
	pthread_mutex_unlock(&pserver->workercnt_mutex);

	return 0;
	
}
